package com.mng.processFun;

import com.mng.dto.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class SideOutPutTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> stringDs = env.socketTextStream("localhost", 7777);
        SingleOutputStreamOperator<SensorReading> sensorDs = stringDs.map(new MapFunction<String, SensorReading>() {
            @Override
            public SensorReading map(String s) throws Exception {
                String[] strings = s.split(",");
                return new SensorReading(strings[0], Long.parseLong(strings[1]), Double.parseDouble(strings[2]));
            }
        });

        OutputTag<SensorReading> lowTempTag = new OutputTag<SensorReading>("lowTemp"){};

        SingleOutputStreamOperator<SensorReading> highSensorDs = sensorDs.keyBy(sensorReading -> sensorReading.getId())
                .process(new KeyedProcessFunction<String, SensorReading, SensorReading>() {
                    @Override
                    public void processElement(SensorReading sensorReading, Context context, Collector<SensorReading> collector) throws Exception {
                        if (sensorReading.getTemperaturre() > 30) {
                            collector.collect(sensorReading);
                        } else {
                            context.output(lowTempTag, sensorReading);
                        }
                    }
                });
        highSensorDs.print("high");
        highSensorDs.getSideOutput(lowTempTag).print("low");
        env.execute();
    }
}
